package com.gradle.quartz.batch.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.gradle.bean.user.User;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.job.builder.FlowBuilder;
import org.springframework.batch.core.job.flow.Flow;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
import org.springframework.batch.item.json.JacksonJsonObjectMarshaller;
import org.springframework.batch.item.json.JacksonJsonObjectReader;
import org.springframework.batch.item.json.JsonFileItemWriter;
import org.springframework.batch.item.json.JsonItemReader;
import org.springframework.batch.item.json.builder.JsonFileItemWriterBuilder;
import org.springframework.batch.item.json.builder.JsonItemReaderBuilder;
import org.springframework.batch.item.validator.BeanValidatingItemProcessor;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.InputStreamResource;
import org.springframework.core.io.PathResource;
import org.springframework.core.task.SimpleAsyncTaskExecutor;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;

@EnableBatchProcessing
@Configuration
public class ConcurrentConfig {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Autowired
    private UseFieldMapper useFieldMapper;
    @Bean
    public JacksonJsonObjectMarshaller<User> jsonObjectMarshaller() {
        return new JacksonJsonObjectMarshaller<>();
    }

    @Bean
    public ItemWriter<User> consoleItemWriter() {
        return items -> {
            for (User item : items) {
                System.out.println(">>>>>>>>>>>>输出到控制台，user = [" + item + "]");
            }
        };
    }

    @Bean
    public FlatFileItemWriter<User> concurrentItemWriter() {
        return new FlatFileItemWriterBuilder<User>()
                .name("userFlatItemWriter")
                .resource(new PathResource("D:\\workspace\\mygradleproject\\readerOut\\readerOut.txt"))
                //指定输出格式
                .formatted()
                .format("id:%s,姓名：%s,年龄：%s,联系地址:%s")
                .names("id", "userName", "age", "address")
                .build();
    }


    @Bean
    public JsonFileItemWriter<User> jsonFileItemWriter() {
        return new JsonFileItemWriterBuilder<User>()
                .name("userJsonItemWriter")
                .resource(new PathResource("D:\\workspace\\mygradleproject\\readerOut\\writerOut.json"))
                .jsonObjectMarshaller(jsonObjectMarshaller())
                .build();
    }


    @Bean
    public FlatFileItemReader<User> concurrentItemReader() {
        return new FlatFileItemReaderBuilder<User>()
                .name("userItemReader")
                .resource(new ClassPathResource("/file/user.txt"))
                //指定解析器，使用#分割，默认是用,号
                .delimited().delimiter("#")
                //截取数据之后，数据怎么命名
                .names("id", "userName", "age", "province", "city", "area")
                //将解析的数据封装到User对象中
//                .targetType(User.class) 自动封装
                .fieldSetMapper(useFieldMapper)
                .build();
    }

    @Bean
    public JsonItemReader<User> jsonItemReader() throws FileNotFoundException {
        ObjectMapper objectMapper = new ObjectMapper();
        JacksonJsonObjectReader<User> userReader = new JacksonJsonObjectReader<>(User.class);
        userReader.setMapper(objectMapper);
        File file = new File("D:\\workspace\\mygradleproject\\gradle-quartz\\src\\main\\resources\\file\\user.json");
        return new JsonItemReaderBuilder<User>()
                .name("userJsonItemReader")
                //放置状态被覆盖
                .saveState(false)
                .resource(new InputStreamResource(new FileInputStream(file)))
                .jsonObjectReader(userReader)
                .build();
    }

    @Bean
    public BeanValidatingItemProcessor<User>concurrentItemProcessor() {
        System.out.println("---------------执行参数校验------------------");
        BeanValidatingItemProcessor<User> processor = new BeanValidatingItemProcessor<>();
        //不满足条件丢弃
        processor.setFilter(true);
        return processor;
    }


    /**
     * 处理任务的逻辑对象
     *
     * @return Tasklett
     */
    @Bean
    public Tasklet concurrentTaskLet() {
        return (contribution, chunkContext) -> {
            System.out.println(">>>>>>>>>jsonTaskLet->处理任务的逻辑对象【jsonTaskLet】>>>>>>>>");
            return RepeatStatus.FINISHED;
        };
    }


    @Bean
    public Step concurrentStep()  {
        System.out.println(">>>>>>>>>concurrentStep-->【concurrentStep】>>>>>>>>");
        return stepBuilderFactory.get("concurrent-step")
                //一次性读多少数据,以及数据类型为User
                .<User, User>chunk(1)
                .reader(concurrentItemReader())
                .writer(concurrentItemWriter())
                .processor(concurrentItemProcessor())
                .build();
    }
    @Bean
    public Step consoleStep() throws FileNotFoundException {
        System.out.println(">>>>>>>>>concurrentStep-->【concurrentStep】>>>>>>>>");
        return stepBuilderFactory.get("concurrent-step")
                //一次性读多少数据,以及数据类型为User
                .<User, User>chunk(1)
                .reader(concurrentItemReader())
                .writer(consoleItemWriter())
                .processor(concurrentItemProcessor())
                .build();
    }

    @Bean
    public Job concurrentJob() throws Exception {
        System.out.println(">>>>>>>>>并行流程【concurrent-flow-job】>>>>>>>>");
        Flow flatFlow = new FlowBuilder<Flow>("concurrentJob")
                .start(concurrentStep())
                .build();
        Flow consoleFlow = new FlowBuilder<Flow>("consoleFlow")
                .start(consoleStep())
                .build();
        Flow jsonFlow = new FlowBuilder<Flow>("jsonFlow")
                .start(concurrentStep())
                .split(new SimpleAsyncTaskExecutor())
                .add(flatFlow,consoleFlow)
                .build();

        System.out.println("thread id" + Thread.currentThread().getId());
        System.out.println("thread 名称" + Thread.currentThread().getName());
        return jobBuilderFactory.get("concurrent-flow-job1")
                .start(jsonFlow)
                .end()
                .build();
    }
}
